package io.kiki.sba.registry.server.metadata.metadataServer.impl;

import io.kiki.sba.registry.exception.InitializeException;
import io.kiki.sba.registry.lifecycle.impl.LifecycleHelper;
import io.kiki.sba.registry.observer.impl.AbstractLifecycleObservable;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.server.metadata.monitor.Metrics;
import io.kiki.sba.registry.store.api.elector.Leader;
import io.kiki.sba.registry.store.api.elector.LeaderAware;
import io.kiki.sba.registry.store.api.elector.LeaderElector;
import io.kiki.sba.registry.store.api.elector.Role;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;


@Component
public class DefaultMetadataLeaderElector extends AbstractLifecycleObservable implements MetadataLeaderService, LeaderAware {

    private final Logger logger = LoggerFactory.getLogger(DefaultMetadataLeaderElector.class);

    private final AtomicBoolean wasLeader = new AtomicBoolean(false);

    private final AtomicReference<LeaderState> leaderState = new AtomicReference<>();

    private final List<LeaderElectorListener> leaderElectorListenerList = new CopyOnWriteArrayList<>();

    @Autowired
    private final LeaderElector leaderElector;

    @Autowired
    private final ServerConfig serverConfig;

    public DefaultMetadataLeaderElector(LeaderElector leaderElector, ServerConfig serverConfig) {
        this.leaderElector = leaderElector;
        this.serverConfig = serverConfig;
    }

    public void registerListener(LeaderElectorListener leaderElectorListener) {
        leaderElectorListenerList.add(leaderElectorListener);
    }

    @PostConstruct
    public void postConstruct() throws Exception {
        LifecycleHelper.initializeIfPossible(this);
        LifecycleHelper.startIfPossible(this);
    }

    @Override
    protected void doInitialize() throws InitializeException {
        super.doInitialize();
        leaderElector.registerLeaderAware(this);
    }

    @Override
    public boolean isWarmuped() {
        return leaderState.get() != null && System.currentTimeMillis() - leaderState.get().getStartTime() > getWarmupMilli();
    }

    @Override
    public boolean amILeader() {
        return leaderElector.amILeader();
    }

    @Override
    public String getLeader() {
        return leaderElector.getLeaderInfo().getLeader();
    }

    @Override
    public long getLeaderEpoch() {
        return leaderElector.getLeaderInfo().getEpoch();
    }

    @Override
    public Leader getLeaderInfo() {
        return leaderElector.getLeaderInfo();
    }

    @Override
    public void leaderNotify() {
        if (wasLeader.compareAndSet(false, true)) {
            leaderState.set(new LeaderState(Role.leader, System.currentTimeMillis()));
            logger.info("[becomeLeader] change from follower to elector, {}", this.leaderState.get());
            leaderElectorListenerList.forEach(LeaderElectorListener::becomeLeader);
        }
    }

    @Override
    public void followNotify() {
        if (wasLeader.compareAndSet(true, false)) {
            leaderState.set(new LeaderState(Role.follower, System.currentTimeMillis()));
            logger.info("[becomeFollow] change from elector to follower, {}", this.leaderState.get());
            // not leader, clear the leader/follower metrics
            Metrics.DataSlot.clearLeaderNumbers();
            Metrics.DataSlot.clearFollowerNumbers();
            leaderElectorListenerList.forEach(LeaderElectorListener::loseLeader);
        }
    }

    private long getWarmupMilli() {
        return serverConfig.getMetaLeaderWarmupMillis();
    }

}
